﻿using Microsoft.Extensions.Logging;

using Quartz.Logging;
using Quartz.Spi;

using LogLevel = Microsoft.Extensions.Logging.LogLevel;

namespace Quartz.Simpl;

/// <summary>
/// An <see cref="IThreadPool"/> implementation which schedules tasks using
/// a <see cref="TaskScheduler"/> (provided by implementers).
/// </summary>
public abstract class TaskSchedulingThreadPool : IThreadPool
{
    private readonly ILogger<TaskSchedulingThreadPool> logger;

    // The token source used to cancel thread pool execution at shutdown
    // Note that cancellation is not propagated to the user-scheduled tasks currently executing,
    // only to the thread pool functions themselves (such as scheduling tasks).
    private readonly CancellationTokenSource shutdownCancellation = new CancellationTokenSource();

    /// <summary>
    /// Allows us to wait until no running tasks remain.
    /// </summary>
    private CountdownEvent runningTasksCountdown = null!;

    /// <summary>
    /// Cached delegate to mark a given task as complete.
    /// </summary>
    private Action<Task> completeTask = null!;

    /// <summary>
    /// The semaphore used to limit concurrency and integers representing maximum
    /// concurrent tasks.
    /// </summary>
    private SemaphoreSlim concurrencySemaphore = null!;

    private int maxConcurrency;
    protected internal const int DefaultMaxConcurrency = 10;

    private TaskScheduler scheduler = null!;
    private bool isInitialized;

    /// <summary>
    /// Gets or sets the <see cref="TaskScheduler"/> used to schedule tasks
    /// queued by users.
    /// </summary>
    /// <remarks>
    /// Once the thread pool is initialized, any attempts to change the value
    /// will be silently ignored.
    /// </remarks>
    public TaskScheduler Scheduler
    {
        get => scheduler;
        set
        {
            if (!isInitialized) scheduler = value;
        }
    }

    /// <summary>
    /// Implementers should override this to provide the TaskScheduler used
    /// by their thread pool.
    /// </summary>
    /// <remarks>
    /// The TaskScheduler is provided through this factory method instead of as a property
    /// so that it can take respect MaxConcurrency changes prior to initialization time
    /// </remarks>
    /// <returns>
    /// The default TaskScheduler the thread pool will use if users do
    /// not specify a different TaskScheduler prior to initialization
    /// </returns>
    protected abstract TaskScheduler GetDefaultScheduler();

    /// <summary>
    /// Gets or sets the maximum number of thread pool tasks which can be
    /// executing in parallel.
    /// </summary>
    /// <remarks>
    /// Once the thread pool is initialized, any attempts to change the value
    /// will be silently ignored.
    /// </remarks>
    public int MaxConcurrency
    {
        get => maxConcurrency;
        set
        {
            if (!isInitialized) maxConcurrency = value;
        }
    }

    /// <summary>
    /// The maximum number of thread pool tasks which can be executing in parallel
    /// </summary>
    /// <remarks>
    /// This alias for MaximumConcurrency is meant to make config files previously used with
    /// SimpleThreadPool or CLRThreadPool work more directly.
    /// </remarks>
    public int ThreadCount
    {
        get => MaxConcurrency;
        set => MaxConcurrency = value;
    }

    // ReSharper disable once UnusedMember.Global
    public string ThreadPriority
    {
        set => logger.LogWarning("Thread priority is no longer supported for thread pool, ignoring");
    }

    /// <summary>
    /// The number of tasks that can run concurrently in this thread pool
    /// </summary>
    public virtual int PoolSize => MaxConcurrency;

    public virtual string InstanceId { get; set; } = null!;

    public virtual string InstanceName { get; set; } = null!;

    protected TaskSchedulingThreadPool() : this(DefaultMaxConcurrency)
    {
    }

    protected TaskSchedulingThreadPool(int maxConcurrency)
    {
        logger = LogProvider.CreateLogger<TaskSchedulingThreadPool>();
        MaxConcurrency = maxConcurrency;
    }

    /// <summary>
    /// Initializes the thread pool for use
    /// </summary>
    /// <remarks>
    /// Note that after invoking this method, neither
    /// </remarks>
    public virtual void Initialize()
    {
        // Checking for null allows users to specify their own scheduler prior to initialization.
        // If this is undesirable, the scheduler should be set here unconditionally.
        if (Scheduler == null)
        {
            Scheduler = GetDefaultScheduler();
        }

        // Initialize the concurrency semaphore with the proper initial count
        concurrencySemaphore = new SemaphoreSlim(MaxConcurrency);

        // We start with an initial count of one to make sure it doesn't start in "signaled" state
        runningTasksCountdown = new CountdownEvent(1);

        // Reduce allocations by caching the delegate to mark a task as complete
        completeTask = SignalTaskComplete;

        // Thread pool is ready to go
        isInitialized = true;

        if (logger.IsEnabled(LogLevel.Debug))
        {
            logger.LogDebug("TaskSchedulingThreadPool configured with max concurrency of {MaxConcurrency} and TaskScheduler {SchedulerName}.",
                MaxConcurrency, Scheduler.GetType().Name);
        }
    }

    /// <summary>
    /// Determines the number of threads that are currently available in
    /// the pool; blocks until at least one is available
    /// </summary>
    /// <returns>The number of currently available threads</returns>
    public int BlockForAvailableThreads()
    {
        if (isInitialized && !shutdownCancellation.IsCancellationRequested)
        {
            try
            {
                // There is a race condition here such that it's possible the method could return
                // 1 (or more) but no threads would be available a short time later when the scheduler
                // calls RunInThread. This could be avoided by 'reserving' threads for callers of
                // BlockForAvailableThreads, but that would complicate this code and nothing should
                // break functionally if threads are used for other tasks in between BlockForAvailableThreads
                // being called and RunInThread being called.
                //
                // The window of opportunity for such a race should be very small (unless the scheduler takes
                // a very long time to call RunInThread).
                //
                // In the worst case, RunInThread will just wait
                // for the next thread and clustered scenarios may experience some imbalanced loads.
                concurrencySemaphore.Wait(shutdownCancellation.Token);
                return 1 + concurrencySemaphore.Release();
            }
            catch (OperationCanceledException)
            {
            }
        }

        return 0;
    }

    /// <summary>
    /// Schedules a task to run (using the task scheduler) as soon as concurrency rules allow it.
    /// </summary>
    /// <param name="runnable">The action to be executed</param>
    /// <returns>
    /// <see langword="true"/> if the task was successfully scheduled; otherwise, <see langword="false"/>.
    /// </returns>
    public bool RunInThread(Func<Task> runnable)
    {
        if (runnable == null || !isInitialized || shutdownCancellation.IsCancellationRequested) return false;

        // Acquire the semaphore (return false if shutdown occurs while waiting)
        try
        {
            concurrencySemaphore.Wait(shutdownCancellation.Token);
        }
        catch (OperationCanceledException)
        {
            return false;
        }

        // Wrap the runnable in a Task to start it asynchronously
        var task = new Task<Task>(runnable);

        // Unrap the task so that we can work with the underlying task
        var unwrappedTask = task.Unwrap();

        lock (runningTasksCountdown)
        {
            // Now that the lock is held, shutdown can't proceed,
            // so double-check that no shutdown has started since the initial check.
            if (shutdownCancellation.IsCancellationRequested)
            {
                concurrencySemaphore.Release();
                return false;
            }

            // Record an extra running task
            runningTasksCountdown.AddCount();
        }

        // Register a callback to remove the task from the running list once it has completed
#pragma warning disable MA0134
        unwrappedTask.ContinueWith(completeTask);
#pragma warning restore MA0134

        // Start the task using the task scheduler
        task.Start(Scheduler);

        return true;
    }

    /// <summary>
    /// Decrements the number of running tasks and releases the concurrency semaphore so that more
    /// tasks may begin running.
    /// </summary>
    /// <param name="completedTask">The task which has completed.</param>
    private void SignalTaskComplete(Task completedTask)
    {
        concurrencySemaphore.Release();
        runningTasksCountdown.Signal();
    }

    /// <summary>
    /// Stops processing new tasks and optionally waits for currently running tasks to finish.
    /// </summary>
    /// <param name="waitForJobsToComplete"><see langword="true"/> to wait for currently executing tasks to finish; otherwise, <see langword="false"/>.</param>
    public void Shutdown(bool waitForJobsToComplete = true)
    {
        logger.LogDebug("Shutting down threadpool...");

        // Cancel using our shutdown token
        shutdownCancellation.Cancel();

        // If waitForJobsToComplete is true, wait for running tasks to complete
        if (waitForJobsToComplete)
        {
            lock (runningTasksCountdown)
            {
                // Cancellation has been signaled, so no new tasks will begin once
                // shutdown has acquired this lock
                logger.LogDebug("Waiting for {ThreadCount} threads to complete.", runningTasksCountdown.CurrentCount.ToString());
            }

            // Signal the initial count that we used to make sure the CountDownEvent didn't start
            // in "signaled" state
            runningTasksCountdown.Signal();

            // Wait for pending tasks to complete
            runningTasksCountdown.Wait();

            logger.LogDebug("No executing jobs remaining, all threads stopped.");
        }

        logger.LogDebug("Shutdown of threadpool complete.");
    }
}